package com.sankuai.xxlrpc.service.impl;

import com.sankuai.xxlrpc.core.model.XxlRpcRegistry;
import com.sankuai.xxlrpc.core.model.XxlRpcRegistryData;
import com.sankuai.xxlrpc.core.model.XxlRpcRegistryMessage;
import com.sankuai.xxlrpc.core.result.ReturnT;
import com.sankuai.xxlrpc.core.util.JacksonUtil;
import com.sankuai.xxlrpc.core.util.PropUtil;
import com.sankuai.xxlrpc.dao.IXxlRpcRegistryDao;
import com.sankuai.xxlrpc.dao.IXxlRpcRegistryDataDao;
import com.sankuai.xxlrpc.dao.IXxlRpcRegistryMessageDao;
import com.sankuai.xxlrpc.service.IXxlRpcRegistryService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.context.request.async.DeferredResult;

import javax.annotation.Resource;
import java.io.File;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Author：YH
 * @Description：
 * @Date:Created in 2022/6/1 12:22
 */
@Service
public class XxlRpcRegistryServiceImpl implements IXxlRpcRegistryService, InitializingBean, DisposableBean {

    private static final Logger logger = LoggerFactory.getLogger(XxlRpcRegistryServiceImpl.class);

    @Resource
    private IXxlRpcRegistryDao xxlRpcRegistryDao;
    @Resource
    private IXxlRpcRegistryDataDao xxlRpcRegistryDataDao;
    @Resource
    private IXxlRpcRegistryMessageDao xxlRpcRegistryMessageDao;

    @Value("${xxl.rpc.registry.data.filepath}")
    private String registryDataFilePath;
    @Value("${xxl.rpc.registry.accessToken}")
    private String accessToken;

    private int registryBeatTime = 10;

    // ------------------------ broadcase + file data ------------------------

    private ExecutorService executorService = Executors.newCachedThreadPool();
    private volatile boolean executorStoped = false;
    private volatile List<Integer> readedMessageIds = Collections.synchronizedList(new ArrayList<Integer>());

    private volatile LinkedBlockingQueue<XxlRpcRegistryData> registryQueue =
            new LinkedBlockingQueue<XxlRpcRegistryData>();
    private volatile LinkedBlockingQueue<XxlRpcRegistryData> removeQueue =
            new LinkedBlockingQueue<XxlRpcRegistryData>();
    private Map<String, List<DeferredResult>> registryDeferredResultMap = new ConcurrentHashMap<>();

    @Override
    public Map<String, Object> pageList(int start, int length, String biz, String env, String key) {
        // page list
        List<XxlRpcRegistry> list = xxlRpcRegistryDao.pageList(start, length, biz, env, key);
        int list_count = xxlRpcRegistryDao.pageListCount(start, length, biz, env, key);

        // package result
        Map<String, Object> maps = new HashMap<String, Object>();
        maps.put("recordsTotal", list_count);        // 总记录数
        maps.put("recordsFiltered", list_count);    // 过滤后的总记录数
        maps.put("data", list);                    // 分页列表
        return maps;
    }

    @Override
    public ReturnT<String> delete(int id) {
        XxlRpcRegistry xxlRpcRegistry = xxlRpcRegistryDao.loadById(id);
        if (xxlRpcRegistry != null) {
            xxlRpcRegistryDao.delete(id);
            xxlRpcRegistryDataDao.deleteData(xxlRpcRegistry.getBiz(), xxlRpcRegistry.getEnv(), xxlRpcRegistry.getKey());

            // sendRegistryDataUpdateMessage (delete)
            xxlRpcRegistry.setData("");
            sendRegistryDataUpdateMessage(xxlRpcRegistry);
        }
        return ReturnT.SUCCESS;
    }

    /**
     * send RegistryData Update Message
     */
    private void sendRegistryDataUpdateMessage(XxlRpcRegistry xxlRpcRegistry) {
        String registryUpdateJson = JacksonUtil.writeValueAsString(xxlRpcRegistry);
        XxlRpcRegistryMessage registryMessage = new XxlRpcRegistryMessage();
        registryMessage.setType(0);
        registryMessage.setData(registryUpdateJson);
        xxlRpcRegistryMessageDao.add(registryMessage);
    }

    @Override
    public ReturnT<String> update(XxlRpcRegistry xxlRpcRegistry) {
        // valid
        if (xxlRpcRegistry.getBiz() == null || xxlRpcRegistry.getBiz().trim().length() < 4 || xxlRpcRegistry.getBiz().trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "业务线格式非法[4~255]");
        }
        if (xxlRpcRegistry.getEnv() == null || xxlRpcRegistry.getEnv().trim().length() < 2 || xxlRpcRegistry.getEnv().trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "环境格式非法[2~255]");
        }
        if (xxlRpcRegistry.getKey() == null || xxlRpcRegistry.getKey().trim().length() < 4 || xxlRpcRegistry.getKey().trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "注册Key格式非法[4~255]");
        }
        if (xxlRpcRegistry.getData() == null || xxlRpcRegistry.getData().trim().length() == 0) {
            xxlRpcRegistry.setData(JacksonUtil.writeValueAsString(new ArrayList<String>()));
        }
        List<String> valueList = JacksonUtil.readValue(xxlRpcRegistry.getData(), List.class);
        if (valueList == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "注册Value数据格式非法；限制为字符串数组JSON格式，如 [address,address2]");
        }

        // valid exist
        XxlRpcRegistry exist = xxlRpcRegistryDao.loadById(xxlRpcRegistry.getId());
        if (exist == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "ID参数非法");
        }

        // if refresh
        boolean needMessage = !xxlRpcRegistry.getData().equals(exist.getData());

        int ret = xxlRpcRegistryDao.update(xxlRpcRegistry);
        needMessage = ret > 0 ? needMessage : false;

        if (needMessage) {
            // sendRegistryDataUpdateMessage (update)
            sendRegistryDataUpdateMessage(xxlRpcRegistry);
        }
        return ret > 0 ? ReturnT.SUCCESS : ReturnT.FAIL;
    }

    @Override
    public ReturnT<String> add(XxlRpcRegistry xxlRpcRegistry) {
        // valid
        if (xxlRpcRegistry.getBiz() == null || xxlRpcRegistry.getBiz().trim().length() < 4 || xxlRpcRegistry.getBiz().trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "业务线格式非法[4~255]");
        }
        if (xxlRpcRegistry.getEnv() == null || xxlRpcRegistry.getEnv().trim().length() < 2 || xxlRpcRegistry.getEnv().trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "环境格式非法[2~255]");
        }
        if (xxlRpcRegistry.getKey() == null || xxlRpcRegistry.getKey().trim().length() < 4 || xxlRpcRegistry.getKey().trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "注册Key格式非法[4~255]");
        }
        if (xxlRpcRegistry.getData() == null || xxlRpcRegistry.getData().trim().length() == 0) {
            xxlRpcRegistry.setData(JacksonUtil.writeValueAsString(new ArrayList<String>()));
        }
        List<String> valueList = JacksonUtil.readValue(xxlRpcRegistry.getData(), List.class);
        if (valueList == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "注册Value数据格式非法；限制为字符串数组JSON格式，如 [address,address2]");
        }

        // valid exist
        XxlRpcRegistry exist = xxlRpcRegistryDao.load(xxlRpcRegistry.getBiz(), xxlRpcRegistry.getEnv(),
                xxlRpcRegistry.getKey());
        if (exist != null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "注册Key请勿重复");
        }

        int ret = xxlRpcRegistryDao.add(xxlRpcRegistry);
        boolean needMessage = ret > 0 ? true : false;

        if (needMessage) {
            // sendRegistryDataUpdateMessage (add)
            sendRegistryDataUpdateMessage(xxlRpcRegistry);
        }
        return ret > 0 ? ReturnT.SUCCESS : ReturnT.FAIL;
    }

    @Override
    public ReturnT<String> registry(String accessToken,
                                    String biz,
                                    String env,
                                    List<XxlRpcRegistryData> registryDataList) {
        // valid
        if (this.accessToken != null && this.accessToken.trim().length() > 0 && !this.accessToken.equals(accessToken)) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "AccessToken Invalid");
        }
        if (biz == null || biz.trim().length() < 4 || biz.trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "Biz Invalid[4~255]");
        }
        if (env == null || env.trim().length() < 2 || env.trim().length() > 255) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "Env Invalid[2~255]");
        }
        if (registryDataList == null || registryDataList.size() == 0) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "Registry DataList Invalid");
        }
        for (XxlRpcRegistryData registryData : registryDataList) {
            if (registryData.getKey() == null || registryData.getKey().trim().length() < 4 || registryData.getKey().trim().length() > 255) {
                return new ReturnT<String>(ReturnT.FAIL_CODE, "Registry Key Invalid[4~255]");
            }
            if (registryData.getValue() == null || registryData.getValue().trim().length() < 4 || registryData.getValue().trim().length() > 255) {
                return new ReturnT<String>(ReturnT.FAIL_CODE, "Registry Value Invalid[4~255]");
            }
        }

        // fill + add queue
        for (XxlRpcRegistryData registryData : registryDataList) {
            registryData.setBiz(biz);
            registryData.setEnv(env);
        }
        registryQueue.addAll(registryDataList);

        return ReturnT.SUCCESS;
    }

    @Override
    public ReturnT<String> remove(String accessToken, String biz, String env,
                                  List<XxlRpcRegistryData> registryDataList) {
        return null;
    }

    @Override
    public ReturnT<Map<String, List<String>>> discovery(String accessToken, String biz, String env, List<String> keys) {
        // valid
        if (this.accessToken != null && this.accessToken.trim().length() > 0 && !this.accessToken.equals(accessToken)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "AccessToken Invalid");
        }
        if (biz == null || biz.trim().length() < 2 || biz.trim().length() > 255) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "Biz Invalid[2~255]");
        }
        if (env == null || env.trim().length() < 2 || env.trim().length() > 255) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "Env Invalid[2~255]");
        }
        if (keys == null || keys.size() == 0) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "keys Invalid.");
        }
        for (String key : keys) {
            if (key == null || key.trim().length() < 4 || key.trim().length() > 255) {
                return new ReturnT<>(ReturnT.FAIL_CODE, "Key Invalid[4~255]");
            }
        }
        Map<String, List<String>> result = new HashMap<String, List<String>>();
        for (String key : keys) {
            XxlRpcRegistryData xxlRpcRegistryData = new XxlRpcRegistryData();
            xxlRpcRegistryData.setBiz(biz);
            xxlRpcRegistryData.setEnv(env);
            xxlRpcRegistryData.setKey(key);

            List<String> dataList = new ArrayList<String>();
            XxlRpcRegistry fileXxlRpcRegistry = getFileRegistryData(xxlRpcRegistryData);
            if (fileXxlRpcRegistry != null) {
                dataList = fileXxlRpcRegistry.getDataList();
            }

            result.put(key, dataList);
        }
        return new ReturnT<Map<String, List<String>>>(result);
    }
    
    // get
    public XxlRpcRegistry getFileRegistryData(XxlRpcRegistryData xxlRpcRegistryData) {

        // fileName
        String fileName = parseRegistryDataFileName(xxlRpcRegistryData.getBiz(), xxlRpcRegistryData.getEnv(),
                xxlRpcRegistryData.getKey());
        // read
        Properties prop = PropUtil.loadProp(fileName);
        if (prop != null) {
            XxlRpcRegistry fileXxlRpcRegistry = new XxlRpcRegistry();
            fileXxlRpcRegistry.setData(prop.getProperty("data"));
            fileXxlRpcRegistry.setStatus(Integer.valueOf(prop.getProperty("status")));
            fileXxlRpcRegistry.setDataList(JacksonUtil.readValue(fileXxlRpcRegistry.getData(), List.class));
            return fileXxlRpcRegistry;
        }
        return null;
    }

    private String parseRegistryDataFileName(String biz, String env, String key) {
        // fileName
        String fileName = registryDataFilePath
                .concat(File.separator).concat(biz)
                .concat(File.separator).concat(env)
                .concat(File.separator).concat(key)
                .concat(".properties");
        return fileName;
    }

    @Override
    public DeferredResult<ReturnT<String>> monitor(String accessToken, String biz, String env, List<String> keys) {
        return null;
    }

    @Override
    public void destroy() throws Exception {
        executorStoped = true;
        executorService.shutdownNow();
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }
}
